using System;
using System.Collections.Concurrent;
using System.Text;
using HslCommunication;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace HslCommunicationEx.RabbitMQ
{
	public class RabbitMqClient
	{
		private ConcurrentDictionary<RmqCommModeEnum, IModel> _dicPublishImodel = new ConcurrentDictionary<RmqCommModeEnum, IModel>();

		private ConcurrentDictionary<RmqCommModeEnum, IModel> _dicSubscribeImodel = new ConcurrentDictionary<RmqCommModeEnum, IModel>();

		private IBasicProperties _ibasicProperties;

		public string HostName { get; set; } = "localhost";


		public string UserName { get; set; }

		public string Password { get; set; }

		public string Uri { get; set; }

		public string VirtualHost { get; set; }

		public int ReconnectInterval { get; set; } = 3000;


		public int RequestConnectionTimeout { get; set; } = 2000;


		public bool? IsPublishOpen(RmqCommModeEnum mode)
		{
			if (!this._dicPublishImodel.ContainsKey(mode))
			{
				return null;
			}
			return this._dicPublishImodel[mode].IsOpen;
		}

		public bool? IsSubscribeOpen(RmqCommModeEnum mode)
		{
			if (!this._dicSubscribeImodel.ContainsKey(mode))
			{
				return null;
			}
			return this._dicSubscribeImodel[mode].IsOpen;
		}

		public OperateResult CheckConnection()
		{
			try
			{
				IConnection connection = this.CreateConnection();
				bool isOpen = connection.IsOpen;
				connection.Close();
				return new OperateResult
				{
					IsSuccess = isOpen
				};
			}
			catch (Exception ex)
			{
				return new OperateResult
				{
					Message = ex.ToString()
				};
			}
		}

		public IConnection CreateConnection()
		{
			try
			{
				ConnectionFactory connectionFactory = null;
				if (!string.IsNullOrWhiteSpace(this.Uri))
				{
					connectionFactory = new ConnectionFactory
					{
						Uri = new Uri(this.Uri)
					};
				}
				else if (!string.IsNullOrWhiteSpace(this.HostName))
				{
					connectionFactory = new ConnectionFactory
					{
						HostName = this.HostName,
						UserName = this.UserName,
						Password = this.Password,
						VirtualHost = this.VirtualHost
					};
				}
				connectionFactory.AutomaticRecoveryEnabled = true;
				connectionFactory.NetworkRecoveryInterval = TimeSpan.FromMilliseconds(this.ReconnectInterval);
				connectionFactory.RequestedConnectionTimeout = TimeSpan.FromMilliseconds(this.RequestConnectionTimeout);
				return connectionFactory.CreateConnection();
			}
			catch (Exception ex)
			{
				throw ex;
			}
		}

		private OperateResult ConsumeMsg(RmqCommModeEnum commMode, string queueName, string exchange, string routingKey, string exchangeType = "topic", bool durable = true, int expirationSecond = 0, bool autoAck = true, Action<object, BasicDeliverEventArgs> rcvFunc = null)
		{
			try
			{
				IModel rcvChannel;
				bool flag = this._dicSubscribeImodel.TryGetValue(commMode, out rcvChannel);
				if (rcvChannel == null)
				{
					IConnection connection = this.CreateConnection();
					rcvChannel = connection.CreateModel();
					this._dicSubscribeImodel.TryAdd(commMode, rcvChannel);
					rcvChannel.BasicQos(0u, 1, global: false);
				}
				Action<IModel, object, BasicDeliverEventArgs> act = null;
				switch (commMode)
				{
				case RmqCommModeEnum.Simple:
					rcvChannel.QueueDeclare(queueName, durable, exclusive: false, autoDelete: false, null);
					act = SimpleRcvMsg;
					break;
				case RmqCommModeEnum.PublishSubscribe:
					act = PublishSubscribeRcvMsg;
					rcvChannel.QueueDeclare(queueName, durable, exclusive: false, autoDelete: false, null);
					rcvChannel.ExchangeDeclare(exchange, exchangeType, durable);
					rcvChannel.QueueBind(queueName, exchange, routingKey);
					break;
				}
				if (durable && this._ibasicProperties == null)
				{
					this._ibasicProperties = this._dicSubscribeImodel[commMode].CreateBasicProperties();
					this._ibasicProperties.Persistent = durable;
					if (expirationSecond > 0)
					{
						this._ibasicProperties.Expiration = (expirationSecond * 1000).ToString();
					}
				}
				EventingBasicConsumer eventingBasicConsumer = new EventingBasicConsumer(rcvChannel);
				eventingBasicConsumer.Received += delegate(object? s, BasicDeliverEventArgs e)
				{
					rcvFunc?.Invoke(s, e);
					act(rcvChannel, s, e);
				};
				rcvChannel.BasicConsume(queueName, autoAck, eventingBasicConsumer);
				return new OperateResult
				{
					IsSuccess = true,
					Message = ""
				};
			}
			catch (Exception ex)
			{
				return new OperateResult
				{
					Message = ex.ToString()
				};
			}
		}

		private OperateResult PublishMsg(RmqCommModeEnum commMode, byte[] body, string queueName, string exchange, string routingKey, string exchangeType = "topic", bool durable = true, int expirationSecond = 0, int waitForConfirmTime = 2000)
		{
			try
			{
				this._dicPublishImodel.TryGetValue(commMode, out var value);
				if (value == null)
				{
					IConnection connection = this.CreateConnection();
					value = connection.CreateModel();
					this._dicPublishImodel.TryAdd(commMode, value);
					switch (commMode)
					{
					case RmqCommModeEnum.Simple:
						value.QueueDeclare(queueName, durable, exclusive: false, autoDelete: false, null);
						break;
					case RmqCommModeEnum.PublishSubscribe:
						value.ExchangeDeclare(exchange, exchangeType, durable);
						break;
					}
					if (durable && this._ibasicProperties == null)
					{
						this._ibasicProperties = this._dicPublishImodel[commMode].CreateBasicProperties();
						this._ibasicProperties.DeliveryMode = 2;
					}
					if (waitForConfirmTime > 0)
					{
						value.ConfirmSelect();
					}
				}
				switch (commMode)
				{
				case RmqCommModeEnum.Simple:
					value.BasicPublish("", queueName, this._ibasicProperties, body);
					break;
				case RmqCommModeEnum.PublishSubscribe:
					value.BasicPublish(exchange, routingKey, this._ibasicProperties, body);
					break;
				}
				bool isSuccess = waitForConfirmTime <= 0 || value.WaitForConfirms(TimeSpan.FromMilliseconds(waitForConfirmTime));
				return new OperateResult
				{
					IsSuccess = isSuccess
				};
			}
			catch (Exception ex)
			{
				return new OperateResult
				{
					Message = ex.ToString()
				};
			}
		}

		public void StartRcvMsg(string queueName, bool autoAck = true, Action<object, BasicDeliverEventArgs> rcvFunc = null)
		{
			this.ConsumeMsg(RmqCommModeEnum.Simple, queueName, null, null, "topic", durable: true, 0, autoAck, rcvFunc);
		}

		public OperateResult SendMsg(string queueName, byte[] body, int waitForConfirmTime = 2000)
		{
			return this.PublishMsg(RmqCommModeEnum.Simple, body, queueName, null, null, "topic", durable: true, 0, waitForConfirmTime);
		}

		public OperateResult SendMsg(string queueName, string exchange, string routingKey, byte[] body, int waitForConfirmTime = 2000)
		{
			return this.PublishMsg(RmqCommModeEnum.Simple, body, queueName, exchange, routingKey, "topic", durable: true, 0, waitForConfirmTime);
		}

		public void SendMsg(string queueName, string body, int waitForConfirmTime = 2000)
		{
			byte[] bytes = Encoding.UTF8.GetBytes(body);
			this.SendMsg(queueName, bytes);
		}

		protected void SimpleRcvMsg(IModel channel, object s, BasicDeliverEventArgs e)
		{
			ReadOnlyMemory<byte> body = e.Body;
			string @string = Encoding.UTF8.GetString(body.ToArray());
			Console.WriteLine("接收到的消息： {0}", @string);
		}

		public OperateResult Subscribe(string queueName, string exchange, string routingKey, string exchangeType = "topic", bool durable = true, int expirationSecond = 0, bool autoAck = true, Action<object, BasicDeliverEventArgs> rcvFunc = null)
		{
			return this.ConsumeMsg(RmqCommModeEnum.PublishSubscribe, queueName, exchange, routingKey, exchangeType, durable, expirationSecond, autoAck, rcvFunc);
		}

		public OperateResult Publish(string queueName, string exchange, string routingKey, byte[] body, string exchangeType = "topic", bool durable = true, int expirationSecond = 0, int waitForConfirmTime = 2000)
		{
			return this.PublishMsg(RmqCommModeEnum.PublishSubscribe, body, queueName, exchange, routingKey, exchangeType, durable, expirationSecond, waitForConfirmTime);
		}

		public OperateResult Publish(string queueName, string exchange, string routingKey, string body, string exchangeType = "topic", bool durable = true, int expirationSecond = 0, int waitForConfirmTime = 2000)
		{
			byte[] bytes = Encoding.UTF8.GetBytes(body);
			return this.PublishMsg(RmqCommModeEnum.PublishSubscribe, bytes, queueName, exchange, routingKey, exchangeType, durable, expirationSecond, waitForConfirmTime);
		}

		protected void PublishSubscribeRcvMsg(IModel channel, object s, BasicDeliverEventArgs e)
		{
			ReadOnlyMemory<byte> body = e.Body;
			string @string = Encoding.UTF8.GetString(body.ToArray());
			Console.WriteLine("接收到的消息： {0}", @string);
		}

		public void Dispose()
		{
			foreach (IModel value in this._dicPublishImodel.Values)
			{
				value?.Dispose();
			}
			this._dicPublishImodel.Clear();
			foreach (IModel value2 in this._dicSubscribeImodel.Values)
			{
				value2?.Dispose();
			}
			this._dicSubscribeImodel.Clear();
		}
	}
}
